package com.mq.java.test;

import com.mq.common.MqConnectionUtils;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeoutException;

/**
 * @author zhd
 * @date 2019/08/17
 *
 * 发送方确认模式有三种，比事物性能好十倍多
 **/
public class ProducerConfirm {

    public static void main(String[] args) {

        try {
            // 连接到RabbitMQ
            Connection connection = MqConnectionUtils.getConnection();

            Channel channel = connection.createChannel();
            System.out.println("信道id = " + channel.getChannelNumber());

            String exchangeName = "confirm-exchange";
            channel.exchangeDeclare(exchangeName,
                    BuiltinExchangeType.DIRECT,
                    false,
                    false,
                    false,
                    null);

            //开启发送方确认模式
            channel.confirmSelect();

            List<Long> messageIdWaitForConfirms = new ArrayList<>(100);
            long id = 100L;
            messageIdWaitForConfirms.add(id);
            channel.basicPublish(exchangeName,
                    "confirm-routingKey",
                    false,
                    false,
                    MessageProperties.TEXT_PLAIN,
                    ("id=" + id + "&age=11").getBytes());

            // 1、等待消息被服务器确认
            if (channel.waitForConfirms()) {
                System.out.println("消息发送成功");
            }

            // 2、同步方式--直到所有信息都发布，只要有一个未确认就会IOException
            channel.waitForConfirmsOrDie();

            // 3、异步监听确认和未确认的消息
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) {
                    System.out.println(String.format("已确认消息，标识：%d，多个消息：%b", deliveryTag, multiple));
                    if (messageIdWaitForConfirms.contains(deliveryTag)) {
                        System.out.println("confirm received");
                        messageIdWaitForConfirms.remove(deliveryTag);
                    }
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) {
                    System.out.println("未确认消息，标识：" + deliveryTag);
                }
            });

            // 渠道关闭
            channel.close();
        } catch (TimeoutException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
